#pragma once
#include <iostream>
#include "buffer.hpp"
#include <condition_variable>
#include <mutex>
#include <atomic>
#include <thread>
#include <functional>

/*   异步工作器
        一种是安全的工作,缓冲区数据满了就等待，一种是不安全:无需等待，缓冲区无限扩容(适用极限性能下)
    对外提供的接口:push数据 1.互斥锁2.是否阻塞 3.放数据 4.唤醒线程
                   threadEntry:线程入口 循环处理 1.互斥锁,读缓冲区是否空，是的话就交换 2.唤醒消费者 3.消费数据 4重置生产缓冲区
                stop()
*/
namespace ns_logger
{
    using Fun = std::function<void(Buffer &)>;
    enum class AsyncSafeType
    {
        Async_Safe = 0,
        Async_Unsafe
    };

    class AsyncLoop
    {
    public:
        AsyncLoop(const Fun &func, AsyncSafeType anynctype = AsyncSafeType::Async_Safe)
            : _safetype(anynctype), _stop(false), _func(func), _thread(std::thread(&AsyncLoop::ThreadEntry, this))
        {
        }
        void Stop()
        {
            _stop = true;
            // 唤醒所有的工作线程
            _cons_cond.notify_all();
            _thread.join();
        }

        ~AsyncLoop()
        {
            Stop();
        }
        void Push(const char *data, size_t len)
        {
            // 往缓冲区内放数据
            // 1.先判断是否为安全放，否则睡眠等待
            // 2.放数据
            // 唤醒消费线程
            std::unique_lock<std::mutex> lock(_mutex);
            if (_safetype == AsyncSafeType::Async_Safe)
                _prod_cond.wait(lock, [&]()
                                { return _prod_buffer.Writeable() >= len; });
            // std::cout<<"异步器push:"<<data<<"_len:"<<len<<std::endl;
            _prod_buffer.Push(data, len);
            _cons_cond.notify_all();
        }

        void ThreadEntry()
        {
            while (1)
            {
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 只有当缓冲区的数据读取完毕&&是stop才break
                    if (_stop && _prod_buffer.Empty())
                        break;
                    //if (_safetype == AsyncSafeType::Async_Safe) // 安全状态下才需要唤醒
                        _cons_cond.wait(lock, [&]()
                                        { return _stop || !_prod_buffer.Empty(); });
                    _cons_buffer.Swap(_prod_buffer);
                    _prod_cond.notify_all();
                }
                // 回调任务处理
                _func(_cons_buffer);
                // 重置缓冲区
                _cons_buffer.ReSet();
            }
        }

    private:
        Fun _func; // 线程处理缓冲区的任务
    private:
        AsyncSafeType _safetype;
        std::atomic<bool> _stop;
        std::mutex _mutex;
        Buffer _prod_buffer;
        Buffer _cons_buffer;
        std::condition_variable _prod_cond;
        std::condition_variable _cons_cond;
        std::thread _thread; // 工作线程
    };

    using AsyncLoopPtr = std::shared_ptr<AsyncLoop>;

};
